package com.atguigu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV2;
import com.atguigu.realtime.bean.VisitorStats;
import com.atguigu.realtime.util.AtguiguUtil;
import com.atguigu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.Map;

import static com.atguigu.realtime.common.Constant.*;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2022/2/16 14:07
 */
public class DwsVisitorApp extends BaseAppV2 {
    public static void main(String[] args) {
        new DwsVisitorApp().init(4001, 1, "DwsVisitorApp", "DwsVisitorApp",
                                 TOPIC_DWD_PAGE, TOPIC_DWM_UV, TOPIC_DWM_UJ_DETAIL);
        
    }
    
    @Override
    protected void run(StreamExecutionEnvironment env,
                       Map<String, DataStreamSource<String>> topicStreamMap) {
        // 1. 解析每个流: 同种类型, 把多个合并成一个流
        DataStream<VisitorStats> visitorStatsStream = parseAndUnionOne(topicStreamMap);
        // 2. 开窗聚合
        SingleOutputStreamOperator<VisitorStats> aggregatedVsStream = aggregate(visitorStatsStream);
        
        // 3. 把结果写入到ClickHouse中
        write2ClickHouse(aggregatedVsStream);
    }
    
    private void write2ClickHouse(SingleOutputStreamOperator<VisitorStats> stream) {
        /*
        自定义ClickHouse Sink
        表可以提前建表.
        只需要执行一个插入的sql语句
        
        在jdbc sink的基础, 封装一个ClickHouse sink
        
        要让类的字段名和表中的字段名保持一致, 主要是为了写一个通用的dao工具
         */
        stream.addSink(FlinkSinkUtil.getClickHouseSink("gmall2022", "visitor_stats_2022", VisitorStats.class));
    }
    
    private SingleOutputStreamOperator<VisitorStats> aggregate(DataStream<VisitorStats> visitorStatsStream) {
        return visitorStatsStream
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    // 加大数据的乱序程度, 防止uv和uj数据迟到
                    .<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                    .withTimestampAssigner((vs, ts) -> vs.getTs())
            )
            .keyBy(vs -> vs.getAr() + "_" + vs.getCh() + "_" + vs.getIs_new() + "_" + vs.getVc())
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .sideOutputLateData(new OutputTag<VisitorStats>("late") {})
            .reduce(
                new ReduceFunction<VisitorStats>() {
                    @Override
                    public VisitorStats reduce(VisitorStats vs1,
                                               VisitorStats vs2) throws Exception {
                        vs1.setPv_ct(vs1.getPv_ct() + vs2.getPv_ct());
                        vs1.setUv_ct(vs1.getUv_ct() + vs2.getUv_ct());
                        vs1.setSv_ct(vs1.getSv_ct() + vs2.getSv_ct());
                        vs1.setUj_ct(vs1.getUj_ct() + vs2.getUj_ct());
                        vs1.setDur_sum(vs1.getDur_sum() + vs2.getDur_sum());
                        
                        return vs1;
                    }
                },
                new WindowFunction<VisitorStats, VisitorStats, String, TimeWindow>() {
                    @Override
                    public void apply(String key,
                                      TimeWindow window,
                                      Iterable<VisitorStats> input,
                                      Collector<VisitorStats> out) throws Exception {
                        VisitorStats vs = input.iterator().next();
                        // 设置窗口开始和结束
                        vs.setStt(AtguiguUtil.toDateTime(window.getStart()));
                        vs.setEdt(AtguiguUtil.toDateTime(window.getEnd()));
                        
                        // 更新统计时间: 设置为系统时间
                        vs.setTs(System.currentTimeMillis());
                        
                        out.collect(vs);
                        
                    }
                }
            );
        
    }
    
    /*
    {
          "common": {
            "ar": "370000",
            "uid": "18",
            "os": "Android 10.0",
            "ch": "xiaomi",
            "is_new": 0,
            "md": "Xiaomi 9",
            "mid": "mid_38",
            "vc": "v2.1.134",
            "ba": "Xiaomi"
          },
          "page": {
            "page_id": "comment",
            "item": "2",
            "during_time": 11791,
            "item_type": "sku_id",
            "last_page_id": "good_spec",
            "source_type": "promotion"
          },
          "ts": 1644637023000
        }
     */
    private DataStream<VisitorStats> parseAndUnionOne(Map<String, DataStreamSource<String>> topicStreamMap) {
        // 1. 用于把计算pv  和 持续访问时长
        SingleOutputStreamOperator<VisitorStats> pvAndDuringTimeStream = topicStreamMap
            .get(TOPIC_DWD_PAGE)
            .map(json -> {
                JSONObject obj = JSON.parseObject(json);
                JSONObject common = obj.getJSONObject("common");
                
                String vc = common.getString("vc");
                String ch = common.getString("ch");
                String ar = common.getString("ar");
                String isNew = common.getString("is_new");
                
                Long ts = obj.getLong("ts");
                
                Long duringTime = obj.getJSONObject("page").getLong("during_time");
                
                return new VisitorStats("", "",
                                        vc, ch, ar, isNew,
                                        0L, 1L, 0L, 0L, duringTime,
                                        ts);
                
            });
        
        // 2. 计算uv
        SingleOutputStreamOperator<VisitorStats> uvStream = topicStreamMap
            .get(TOPIC_DWM_UV)
            .map(json -> {
                JSONObject obj = JSON.parseObject(json);
                JSONObject common = obj.getJSONObject("common");
                
                String vc = common.getString("vc");
                String ch = common.getString("ch");
                String ar = common.getString("ar");
                String isNew = common.getString("is_new");
                
                Long ts = obj.getLong("ts");
                
                return new VisitorStats("", "",
                                        vc, ch, ar, isNew,
                                        1L, 0L, 0L, 0L, 0L,
                                        ts);
                
            });
        
        // 3. 计算sv  进入次数,用于计算跳出率的分母的
        SingleOutputStreamOperator<VisitorStats> svStream = topicStreamMap
            .get(TOPIC_DWD_PAGE)
            .flatMap(new FlatMapFunction<String, VisitorStats>() {
                @Override
                public void flatMap(String json,
                                    Collector<VisitorStats> out) throws Exception {
                    
                    JSONObject obj = JSON.parseObject(json);
                    // 判断是否被进入
                    String lastPageId = obj.getJSONObject("page").getString("last_page_id");
                    if (lastPageId == null || lastPageId.length() == 0) {
                        JSONObject common = obj.getJSONObject("common");
                        
                        String vc = common.getString("vc");
                        String ch = common.getString("ch");
                        String ar = common.getString("ar");
                        String isNew = common.getString("is_new");
                        
                        Long ts = obj.getLong("ts");
                        VisitorStats vs = new VisitorStats("", "",
                                                           vc, ch, ar, isNew,
                                                           0L, 0L, 1L, 0L, 0L,
                                                           ts);
                        out.collect(vs);
                    }
                    
                }
            });
        // 4. 计算uj  跳出次数
        SingleOutputStreamOperator<VisitorStats> ujStream = topicStreamMap
            .get(TOPIC_DWM_UJ_DETAIL)
            .map(json -> {
                JSONObject obj = JSON.parseObject(json);
                JSONObject common = obj.getJSONObject("common");
                
                String vc = common.getString("vc");
                String ch = common.getString("ch");
                String ar = common.getString("ar");
                String isNew = common.getString("is_new");
                
                Long ts = obj.getLong("ts");
                
                return new VisitorStats("", "",
                                        vc, ch, ar, isNew,
                                        0L, 0L, 0L, 1L, 0L,
                                        ts);
                
            });
        
        return pvAndDuringTimeStream.union(uvStream, svStream, ujStream);
        
    }
}
/*
{
  "common": {
    "ar": "370000",
    "uid": "18",
    "os": "Android 10.0",
    "ch": "xiaomi",
    "is_new": 0,
    "md": "Xiaomi 9",
    "mid": "mid_38",
    "vc": "v2.1.134",
    "ba": "Xiaomi"
  },
  "page": {
    "page_id": "comment",
    "item": "2",
    "during_time": 11791,
    "item_type": "sku_id",
    "last_page_id": "good_spec",
    "source_type": "promotion"
  },
  "ts": 1644637023000
}


uv和uj数据迟到的两种解决方案:
1. 增加乱序程度

2. 在union之前设置水印, 那么开窗的时候, 以更新笔记满的水印位置(uj或uv)
 */
